package com.jch.d1;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

public class source_UDF {

    public static void main(String[] args) throws Exception{

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //添加自定义数据源
        DataStreamSource source = env.addSource(new MySource());

        //打印
        source.print();

        //执行
        env.execute();


    }

    //自定义source
    public static class MySource implements SourceFunction{

        private boolean running=true;
        @Override
        public void run(SourceContext sourceContext) throws Exception {
            //生成随机数
            Random random = new Random();

            //顶一个map集合
            HashMap<String, Double> map = new HashMap<String, Double>();

            //循环赋值
            for(int i=0;i<10;i++){
                map.put("sender_"+(i+1),60+random.nextGaussian()*20);
            }

            //死循环
            while (running){
                for(String id:map.keySet()){
                    Double gaussian = random.nextGaussian();
                    sourceContext.collect(new Sourcedata(id,System.currentTimeMillis(),gaussian));
                }
            }
        }

        @Override
        public void cancel() {
           running=false;
        }
    }




}
